package com.lujianfei.androidthread.threadsActivity;

import androidx.appcompat.app.AppCompatActivity;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observers.DisposableObserver;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;

import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.TextView;

import com.lujianfei.androidthread.R;

public class RxJavaActivity extends AppCompatActivity implements View.OnClickListener {

    private static final String TAG = "RxJavaActivity==";
    private TextView txt_info;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_java);
        RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.e(TAG, "setErrorHandler accept " + throwable);
            }
        });
        txt_info = findViewById(R.id.txt_info);
        findViewById(R.id.bt_clickme).setOnClickListener(this);
    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.bt_clickme: {
                processClickMe();
            }
            break;
        }
    }

    private void processClickMe() {

        //创建两个上游, 被观察者 Register, Login，实现注册后登录的操作 ：
        final Observable observableRegister = Observable.create(new RegisterObservableOnSubscribe());
        final Observable observableLogin = Observable.create(new LoginObservableOnSubstribe());

        observableRegister
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .flatMap(new Function<Boolean,Observable<Boolean>>() {
                    @Override
                    public Observable apply(Boolean o) throws Exception {
                        return observableLogin;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new DisposableObserver() {
                    //创建一个下游  观察者Observer
                    @Override
                    protected void onStart() {
                        super.onStart();
                        // 这里是主线程 (UI线程) ，执行顺序：最开始
                        Log.d(TAG, "observer onStart thread is : " + Thread.currentThread().getName());
                        txt_info.setText("Loading ...");
                    }

                    @Override
                    public void onNext(Object o) {
                        // 这里是主线程 (UI线程) ，执行顺序：最后
                        Log.d(TAG, "observer onNext thread is : " + Thread.currentThread().getName());
                        txt_info.setText("Finish");
                    }

                    @Override
                    public void onError(Throwable e) {
                        // 这里是主线程 (UI线程) ，执行顺序：最后
                        Log.d(TAG, "observer onError thread is : " + Thread.currentThread().getName());
                        txt_info.setText("onError " + e);
                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    class RegisterObservableOnSubscribe implements ObservableOnSubscribe<Boolean>{

        @Override
        public void subscribe(ObservableEmitter<Boolean> emitter) throws Exception {
            // 这里是子线程 ，执行顺序：2
            Log.d(TAG, "Register subscribe thread is : " + Thread.currentThread().getName());
            Thread.sleep(1000);
            emitter.onNext(true); // 注册成功
            // emitter.onError(new Exception("注册失败")); // 注册失败
        }
    }

    class LoginObservableOnSubstribe implements ObservableOnSubscribe<Boolean>{

        @Override
        public void subscribe(ObservableEmitter<Boolean> emitter) throws Exception {
            // 这里是子线程 ，执行顺序：3
            Log.d(TAG, "Login subscribe thread is : " + Thread.currentThread().getName());
            Thread.sleep(1000);
            emitter.onNext(true); // 登录成功
            // emitter.onError(new Exception("登录失败")); // 登录失败
        }
    }
}
